/* $Id: channel.cc jjs $ */
/*
   Copyright (C) 2007 by Jurjen Stellingwerff <jurjen@stwerff.xs4all.nl>
   Part of the Moros Project http://moros.sourceforge.net/

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License version 2 or any
   later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY.

   See the COPYING file for more details.
*/
#define DEBUG_ROUTINE "channel"

#include "channel.h"
#include "util.h"
#include "log.h"
#include "misc.h"
#include "communicate.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <iostream>
#include <errno.h>
#include <sstream>

// import only needed for game_server
#include "import.h"

#define COMM_TIMEOUT 1800
#define min(a,b) (((a)<(b))?(a):(b))

/// Write the length data that would be transferred to the console

class debug_conn: public connection {
private:
public:
	debug_conn() {}
	int read(char *buffer, size length) {
		dbg_comm("Reading from debug channel length %u", length);
		return 0;
	}
	int write(const char *buffer, size length) {
		dbg_comm("Writing to debug channel length %u", length);
		return 0;
	}
	void close() {
	    dbg_comm("Close debug channel");
	}
	size length() {
		return 0;
	}
};

void channel::send_session(long time) {
	last_=time;
	std::string send="user\t"+comm->user+"\nsession\t"+derived(comm->session, comm->sessnr)+"\n";
	char len[9];
	unsigned long int s=send.length();
	sprintf(len, "%8li", s);
	len[8]=0;
	send=len+send;
	int bc=send.length();
	while (bc>0) {
		int rc=stream->write(send.c_str(), bc);
		if (rc<0 && errno!=EAGAIN) {
			dbg_error("Could not write session ID");
			return;
		}
		if (rc>0) bc-=rc;
	}
	comm->sessnr++;
}

bool channel::check_session() {
	if (key=="session") {
		if (derived(comm->session, comm->sessnr) == value) {
			dbg_state("Connected correctly to the central server");
			comm->sessnr++;
			return true;
		}
        dbg_error("Didn't got a correct session id from the server");
		comm->sessnr--;
        finish();
        return false;
	} else {
	    dbg_error("Expecting session but found %s with %s", key.c_str(), value.c_str());
	}
	return false;
}

channel::channel(int channeltype, long time) {
	input_ = new buffer();
	output_ = new buffer();
	type=channeltype;
	if (type==TYPE_SOCKET) {
        output_ -> write("00000000");
		to_input_ = 8;
		state = STATE_WAITING;
	} else {
		to_input_ = -1;
		state = STATE_READING;
	}
	stream = new debug_conn();
	last_ = time;
	finished = false;
	lock = false;
	comm = 0;
	readf = -1;
	writef = -1;
	errors = 0;
}

bool channel::peek(string key) {
    string cmp=key.append("\t");
	char tmp[cmp.length()+1];
	strncpy(tmp, input_->ptr(), min(cmp.length(), input_->bytes()));
	tmp[cmp.length()]=0;
	if (input_->bytes() > cmp.length()) dbg_calls("Peek if '%s' is '%s'", tmp, cmp.c_str());
	else dbg_calls("Peek: stream ended before '%s'", cmp.c_str());
	return (cmp.compare(tmp)==0);
}

bool channel::next(void) {
    if (type==TYPE_IMPORT) return import_next(this);
	char *str=input_->ptr();
	long max=input_->bytes();
	if (max==0 || finished) {
	    key="";
	    value="";
		return false;
	}
	long int pos=0;
	while(str[pos]!=9 && str[pos]!=13 && str[pos]!=0 && pos<max) pos++;
	str[pos]=0;
	key=str;
	pos++;
	while(str[pos]==9 && pos<max) pos++;
	int vpos=pos;
	while(str[pos]!=10 && str[pos]!=0 && pos<max) {
		if (str[pos]==1) str[pos]=10;
		pos++;
	}
	str[pos]=0;
	value=str+vpos;
	dbg_data("Next key is '%s' with value '%s'", str, value.c_str());
	input_->done(pos+1);
	return true;
}

void channel::write(const std::string key, const std::string value) {
    dbg_calls("Write key '%s' with value '%s'", key.c_str(), value.c_str());
	int pos=key.length();
	int len=key.length()+1+value.length()+1;
	char *str=(char*)malloc(len+1);
	strcpy(str, key.c_str());
	str[pos]=9;
	strcpy(str+pos+1, value.c_str());
	for(int i=pos+1; i<len-1; i++) {
		if (str[i]==10) str[i]=1;
	}
	str[len-1]=10;
	str[len]=0;
	if (type == TYPE_FILE) {
		stream->write(str, len);
	} else {
		output_->write(str);
	}
}

void channel::write(const std::string key, const long value) {
	std::ostringstream streamOut;
    streamOut << value;
    write(key, streamOut.str());
}

void channel::finish() {
	if (type == TYPE_FILE) {
		stream->close();
	}
	state = STATE_CLOSED;
	finished = true;
	dbg_comm("Set channel %i on closed", id);
}

void channel::close(void) {
	input_->~buffer();
	output_->~buffer();
	stream->close();
}

void channel::send() {
	long int length;
	char * ptr=output_->send(&length);
	char tmp[9];
	snprintf(tmp, 9, "%08li",  length-8);
	memcpy(ptr,tmp,8);
	output_ -> write("00000000");
	//type=STATE_WRITING;
}

void channel::fread(const std::string file) {
    dbg_calls("Reading data from file: %s", file.c_str());
    if (type==TYPE_IMPORT) {
        set_file(file);
        return;
    }
    readf=open(file.c_str(), O_RDONLY);
    int l=0;
    while(l==0) {
        l=output_->fwrite(readf);
    }
}

void channel::fwrite(const std::string file) {
    if (writef!=-1) {
        write_files.push_back(file);
        dbg_calls("Remember file %s to write to", file.c_str());
    } else {
        dbg_calls("Start writing data to file %s", file.c_str());
        writef=open(file.c_str(), O_WRONLY | O_CREAT | O_TRUNC);
    }
}

void channel::file_poll(long time) {
	int rc;
	if (state == STATE_WAITING) {
		state = STATE_READING;
		to_input_=-1;
	}
	if (state == STATE_READING) {
		if (to_input_<0) {
			to_input_=stream->length();
			dbg_calls("Poll file with %u bytes", to_input_);
		}
		if (to_input_ >0) {
			char *ptr=input_->writing(to_input_);
			rc=stream->read(ptr, to_input_);
			if (rc>=0) {
				to_input_ -= rc;
				last_ = time;
				input_->wrote(rc);
			} else {
			    errors++;
			}
		}
		if (to_input_==0) {
			dbg_comm("Finished reading file");
			input_->send(NULL);
			if (next()) handler.emit(this);
			if (input_->bytes()>0) dbg_error("Throwing away %u bytes", input_->bytes());
			input_->done(input_->bytes());
			state = STATE_CLOSED;
			finished=true;
		}
	}
}

/// Scan this channel for input
/**
\todo Better check on incorrect data\n
    More gracious handling when the number of bytes of the data is no real number
    - dump current data from the stream... so we hope that it cleans the stream for the next data
    - good logging of these type of events... there is always data loss
*/

void channel::poll(long time) { // add state
    int rc;
	if (type == TYPE_FILE) {
	    file_poll(time);
	    return;
	}
	if (state == STATE_CLOSED && comm!=NULL && comm->session!="" && output_->bytes()>0) {
		state = STATE_WAITING;
		comm->reconnect(this);
		dbg_calls("Try to reconnect");
	}
	if (time > last_+ COMM_TIMEOUT) {
		state = STATE_CLOSED;
		finished = true;
		dbg_comm("Timeout on channel %i", id);
	}
	if (errors>15) {
		state = STATE_CLOSED;
		finished = true;
		dbg_comm("Too many errors on channel %i", id);
	}
	if (state == STATE_WAITING) {
		if (to_input_ > 0) {
			char *ptr=input_->writing(to_input_);
			rc=stream->read(ptr, to_input_);
			if (rc>0) {
				last_ = time;
				if (errors>0) errors--;
				input_->wrote(rc);
				to_input_ -= rc;
				dbg_calls("Read %i bytes of %li", rc, to_input_);
			} else if (rc<0) {
			    errors++;
			    //dbg_calls("Increased errors to %i",errors);
			}
		}
		if (to_input_ == 0) {
			char tmp[9];
			char *ptr=input_->ptr();
			memcpy(tmp, ptr, 8);
			tmp[8]=0;
			to_input_ = toint(tmp);
            state = STATE_READING;
		}
	}
	if (state == STATE_WAITING && to_input_==8 && output_->bytes()>0) {
		state = STATE_WRITING;
	}
    if (state == STATE_READING) {
        if (to_input_ > 0) {
            char *ptr=input_->writing(to_input_);
            rc=stream->read(ptr, to_input_);
            if (rc>=0) {
                to_input_ -= rc;
                if (errors>0) errors--;
                last_ = time;
                input_->wrote(rc);
            } else {
			    errors++;
			    //dbg_calls("Increased errors to %i",errors);
			}
            if (to_input_ > 0) {
                dbg_comm("Still to read %i bytes of %i total from channel %i", to_input_, rc, id);
            }
        }
        if (to_input_ == 0) {
            input_->send(NULL);
            input_->done(8);
            if (lock && type==TYPE_SOCKET) {
                dbg_calls("Testing for session on locked channel %i", id);
                if (next()) {
                    if (check_session()) {
                        lock=false;
                    } else {
                        finish();
                    }
                }
            }
            if ((!lock || type==TYPE_SERVER) && writef>-1) {
                dbg_calls("Writing %li bytes to file", input_->bytes());
                ::write(writef, input_->ptr(), input_->bytes());
                input_->done(input_->bytes());
                if (!write_files.empty()) {
                    string file=write_files.front();
                    write_files.pop_front();
                    dbg_calls("Start writing to file %s now", file.c_str());
                    writef=open(file.c_str(), O_WRONLY | O_CREAT | O_TRUNC);
                } else writef=-1;
            } else {
                if ((!lock || type==TYPE_SERVER) && next()) handler.emit(this);
                if (input_->bytes()>0) dbg_comm("Throwing away %i bytes on channel %i", input_->bytes());
                input_->done(input_->bytes());
            }
            if (output_->bytes() > 0) {
                state = STATE_WRITING;
            } else {
                state = STATE_WAITING;
                to_input_ =8;
            }
        }
    }
	if (state == STATE_WRITING) {
		int bc=output_->bytes();
		if (bc>0) {
			char *ptr=output_->ptr();
			int err=ECONNRESET;
			while (err==ECONNRESET) {
				rc=stream->write(ptr, bc);
				if (rc>0) {
					err=0;
					last_ = time;
					if (errors>0) errors--;
					output_->done(bc);
					if (rc==bc) {
						to_input_=8;
						state = STATE_WAITING;
					} else {
						dbg_comm("Still to send %i bytes of %i total from channel %i", bc-rc, bc, id);
					}
				} else if (comm->session!="" && rc<0 && (errno==ECONNABORTED || errno==ECONNRESET)) {
				    errors++;
					err=errno;
					if (comm->reconnect(this)) {
						err=0;
					}
				} else if (rc<0) {
                    errors++;
                    //dbg_calls("Increased errors to %i",errors);
                }
			}
		} else {
			to_input_=8;
			state = STATE_WAITING;
		}
	}
}
